package biz.ws.ws;

import api.common.IdKey;
import api.common.bean.SysUser;
import api.common.result.Result;
import api.common.result.SystemException;
import api.common.service.dubbo.RemoteAuthService;
import biz.ws.WsConfig;
import biz.ws.receive.MessageExecutorContainer;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.function.Consumer;

/**
 * @创建人 wsdc
 * @时间 2021/4/1
 * @描述 消息统一处理
 */
@Component
@Slf4j
public class WebsocketServiceImpl implements WebsocketService{
    @Resource
    UserContainer container;

    @Resource
    MessageExecutorContainer messageHandlerContainer;

    @Resource
    MsgDataSource msgDataSource;

    @Resource
    WsConfig wsConfig;

    @DubboReference(check = false)
    RemoteAuthService remoteAuthService;

    @Resource
    IdKey idKey;


    @Override
    public void onMessageFromClient(MsgHolder context) {
        Channel channel = context.getCtx().channel();
        Message message = context.getMessage();

        // TODO: 2021/4/4 参数校验
        if(!WsCst.MsgType.check(message.getType())){
            //  不支持的访问数据拒绝掉
            log.info("WebsocketServiceImpl onMessageFromClient 不支持的消息类型 message= {}",JSON.toJSONString(message));
            return ;
        }

        MsgSession session = container.get(channel);
        //  如果session不存在 此时是不接受消息的 没有登录校验之前接收到的消息
        if(session == null){
            return ;
        }

        // TODO: 2021/4/4 分配id  参数校验 ????
        message.setId(acquireId());
        message.setFromId(session.getUser().getUserId());
        message.setCreateTime(LocalDateTime.now());
        //  分发给具体的处理去处理
        messageHandlerContainer.onClientMessage(context);
        log.info("WebsocketServiceImpl onMessage message = {}",JSON.toJSONString(context.getMessage()));
    }

    @Override
    public void onMessageFromKafka(MsgHolder context) {
        // TODO: 2021/4/4 参数校验
        Message message = context.getMessage();
        if(!WsCst.MsgType.check(message.getType())){
            //  不支持的访问数据拒绝掉
            log.info("WebsocketServiceImpl onMessageFromClient 不支持的消息类型 message= {}",JSON.toJSONString(message));
            return ;
        }
        messageHandlerContainer.onClientMessage(context);
        log.info("WebsocketServiceImpl onMessage message = {}",JSON.toJSONString(context.getMessage()));
    }

    @Override
    public void onLogin(MsgHolder context) {
        WsModel.LoginInfo loginInfo = context.getLoginInfo();

        log.info("WebsocketServiceImpl onLogin login = {}", JSON.toJSONString(loginInfo));

        MsgSession session = new MsgSession();

        login0(context,user -> {
            session.setUser(user);
            session.setLoginFrom(loginInfo.getLoginFrom());

            Channel channel = context.getCtx().channel();
            Channel before = container.put(session, channel);

            //  关闭之前已经存在的
            if(before != null){
                if(before.isActive()){
                    WsUtils.close(before);
                }
            }

            //  查询历史没有发送的信息
            msgDataSource.loadUnreadMsg(LocalDateTime.now().plusMonths(-1), user.getUserId(), list -> {
                for (Message message : list) {
                    WsUtils.write(channel,message);
                }
                //  将没有发送的消息推送 即可  客户端收到回复在更新状态
                //msgDataSource.update(list);
            });
        });

    }

    //  具体获取用户的方式
    protected void login0(MsgHolder context, Consumer<SysUser> consumer){
        switch (wsConfig.getLoginType()){
            case WsCst.LoginType.DIRECT:
            {
                SysUser user = new SysUser();
                WsModel.LoginInfo loginInfo = context.getLoginInfo();
                user.setUserId(Long.parseLong(loginInfo.getUserId()));
                consumer.accept(user);
            }
                break;

            case WsCst.LoginType.PASSWORD:
            {
                // TODO: 2021/4/4 密码
            }
                break;

            //  token 的校验方式
            case WsCst.LoginType.TOKEN:
            {
                Result<SysUser> userByToken = remoteAuthService.getUserByToken(context.getLoginInfo().getToken());
                if(userByToken.code() == 0){
                    consumer.accept(userByToken.data());
                    return ;
                }
                throw new SystemException(userByToken);
            }
        }

        return ;
    }

    @Override
    public void onMessageTypeNotSupport(MsgHolder context) {
        //  直接退出
        WsUtils.close(context.getCtx().channel());
        log.info("WebsocketServiceImpl onMessageTypeNotSupport message = ",JSON.toJSONString(context.getMessage()));
    }

    @Override
    public void onException(MsgHolder context) {
        log.info("WebsocketServiceImpl onException e= ",context.getE());
    }

    @Override
    public void onUnregister(MsgHolder context) {
        Channel channel = context.getCtx().channel();
        container.removeConnection(channel);
        log.info("WebsocketServiceImpl onUnregister 退出登录 ");
    }

    @Override
    public void onStart(ChannelFuture future) {
        log.info("WebsocketServiceImpl onStart ws 服务启动成功");
    }

    @Override
    public void onConnect(MsgHolder holder) {
        if(container.size() >= wsConfig.getMaxConnect()){
            Channel channel = holder.getCtx().channel();
            WsUtils.close(channel,new Result().code(-1).msg("系统超负载"));
        }
    }

    @Override
    public void onNotBeat(MsgHolder context) {
        WsUtils.write(context.getCtx().channel(),new Result().code(-1).msg("需要心跳"));
        log.info("WebsocketServiceImpl onNotBeat 长时间没有发送心跳 是否关闭");
    }

    private long acquireId(){
        return idKey.longId();
    }
}
